/*
 ***************************************************************************************
 * All rights Reserved, Designed By RTF
 * @Title:  KafkaProducerAdaptor.java   
 * @Package com.rtf.repair.mq.kafka.producer
 * @Description: kafka 生产者适配器 
 * @author: 司福林
 * @date:   2020-10-15 19:57:58   
 * @version V1.0 
 * @Copyright: 2020 RTF. All rights reserved.
 * 注意：本内容仅限于公司内部使用，禁止外泄以及用于其他的商业目
 *  ---------------------------------------------------------------------------------- 
 * 文件修改记录
 *     文件版本：         修改人：             修改原因：
 ***************************************************************************************
 */
package com.rtf.framework.mq.kafka.producer;

import java.io.Serializable;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.util.Assert;

import com.rtf.framework.mq.common.CallBack;
import com.rtf.framework.mq.common.MessageProducer;
import com.rtf.framework.mq.utils.ByteUtils;
import com.google.common.collect.Lists;

/**
 *  
 * @Description: kafka 生产者适配器
 * @ClassName:  KafkaProducerAdaptor
 * @author: 67077
 * @date:   2020-10-15 19:57:58
 * @since:  v1.0
 */
public class KafkaProducerAdaptor implements MessageProducer {

	private Producer<String, byte[]> producer;

	public KafkaProducerAdaptor(Producer<String, byte[]> producer) {
		Assert.notNull(producer, "Kafka producer is null");

		this.producer = producer;
	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg) {
		doSend(buildRecord(topic, null, key, msg, null));

	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, CallBack callBack) {
		doSend(buildRecord(topic, null, key, msg, null), callBack);

	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, Integer parttion) {
		doSend(buildRecord(topic, null, key, msg, parttion));
	}

	@Override
	public void sendMsg(String topic, String key, Serializable msg, Integer parttion, CallBack callBack) {
		doSend(buildRecord(topic, null, key, msg, parttion), callBack);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg) {
		doSend(buildRecord(topic, tag, key, msg, null));

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, CallBack callBack) {
		doSend(buildRecord(topic, tag, key, msg, null), callBack);

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, Integer parttion) {
		doSend(buildRecord(topic, tag, key, msg, parttion));

	}

	@Override
	public void sendMsg(String topic, String tag, String key, Serializable msg, Integer parttion, CallBack callBack) {
		doSend(buildRecord(topic, tag, key, msg, parttion), callBack);

	}

	private ProducerRecord<String, byte[]> buildRecord(String topic, String tag, String key, Serializable msg, Integer parttion) {
		Iterable<Header> headers = null;
		if (!StringUtils.isBlank(tag)) {
			List<Header> headList = Lists.newArrayList();
			headList.add(new RecordHeader(MessageConst.PROPERTY_TAGS, tag.getBytes()));
			headers = (Iterable<Header>) headList.iterator();
		}
		ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(topic, parttion, key, ByteUtils.toBytes(msg), headers);
		return record;

	}

	private void doSend(ProducerRecord<String, byte[]> rcord) {
		producer.send(rcord);
	}

	private void doSend(ProducerRecord<String, byte[]> rcord, CallBack callBack) {
		producer.send(rcord, new KafkaProducerCallback(callBack));
	}

	@Override
	public Object getProducer() {
		// TODO Auto-generated method stub
		return this.producer;
	}

}
